package com.walala.dispatcher.zk;

import java.util.List;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkClient {

	private static Logger logger = LoggerFactory.getLogger(ZkClient.class);

	private static CuratorFramework client;
	
	private static CountDownLatch LATCH = new CountDownLatch(1);

	private static ZkClient INSTANCE = new ZkClient();
	
	private static ConcurrentSkipListSet<Boolean> CONTAINS_SET = new ConcurrentSkipListSet<>();

	public static ZkClient getInstance() {
		return INSTANCE;
	}

	public void initZkClient(String zkIpPort) {
		client = CuratorFrameworkFactory.builder().connectString(zkIpPort)
				// .namespace(namespace)
				.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 2000)).build();
		// 连接状态监听
		ConnectionStateListener listener = new ConnectionStateListener() {
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				if (newState == ConnectionState.CONNECTED) {
					logger.info("ZK连接成功");
					LATCH.countDown();
				}
			}
		};
		client.getConnectionStateListenable().addListener(listener);
		client.start();

	}
	
	public void getChildPath(final String rootPath, final ConcurrentSkipListSet<String> serverList){
		
		try {
			Watcher watcher = new Watcher() {
				
				@Override
				public void process(WatchedEvent event) {
					if(Event.EventType.NodeChildrenChanged == event.getType()){
						logger.info("the child of parentPath[{}] has changed!",rootPath);
						try {
							List<String> childrenPaths = client.getChildren().forPath(rootPath);
							serverList.addAll(childrenPaths);
		                    serverList.retainAll(childrenPaths);
						} catch (Exception e) {
							logger.error("watcher child state of parentPath[{}] error",rootPath,e);
						}
	                    
					}
				}
			};
		
			List<String> childrenPaths = client.getChildren().usingWatcher(watcher).forPath(rootPath);
			serverList.addAll(childrenPaths);
			
		} catch (Exception e) {
			logger.error("getChildPath of parentPath[{}] error",rootPath,e);
		}
	}

	private static void checkClientState() {
		
	}

	public void startPathChildListen(String path){
		if(client == null){
			logger.error("zkClient is not start,please start the client first!");
			return;
		}
		CuratorFrameworkState state = client.getState();
		if(state.STARTED != state){
			logger.error("zkClient is not start,please start the client first!");
			return;
		}
		// 某个路径下节点监听
		try {
			PathChildrenCache pathWatcher = new PathChildrenCache(client,path,true);
			pathWatcher.start();
			pathWatcher.getListenable().addListener(new PathChildrenCacheListener() {

				@Override
				public void childEvent(CuratorFramework client_, PathChildrenCacheEvent event) throws Exception {
					ChildData childData = event.getData();
					if (childData == null) {
						logger.info("No data in event:{}", event);
					} else {
						Type type = event.getType();
						String path = childData.getPath();
						byte[] data = childData.getData();
						String data_ = new String(data);
						Stat stat = childData.getStat();
						logger.info("节点发生变化，路劲：{}，类型：{}，数据：{}，状态：{}", path, type, data_, stat);
					}

				}
			});
		} catch (Exception e) {
			logger.error("启动监听某个路径下节点变化异常", e);
		}
	}

	public void createRootNode(String rootPath) {
		try {
			Stat state = client.checkExists().forPath(rootPath);
			if (state == null) {
				client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(rootPath);
			} else {
				logger.info("根节点已经存在,ROOT_PATH:{}", rootPath);
			}
		} catch (Exception e) {
			logger.error("error:{}", e);
		}
	}
	
	public void createNodeAndData(String path,String data){
		try {
			Stat state = client.checkExists().forPath(path);
			if (state == null) {
				client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT);
				if(data == null){
					client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
				}else{
					client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, data.getBytes());
				}
			} else {
				logger.info("节点已经存在,path:{}", path);
			}
		} catch (Exception e) {
			logger.error("创建节点或数据错误",e);
		}
	}
	
	public void createNode(String path,String data,boolean isPersistent){
		try {
			Stat stat = client.checkExists().forPath(path);
			if(stat == null){
				if(isPersistent){
					if(StringUtils.isBlank(data)){
						client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
					}else{
						client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, data.getBytes());
					}
				}else{
					if(StringUtils.isBlank(data)){
						client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
					}else{
						client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, data.getBytes());
					}
				}
			}else{
				if(StringUtils.isBlank(data)){
					logger.info("节点已存在,path:[{}]",path);
				}else{
					client.setData().forPath(path, data.getBytes());
				}
			}
		} catch (Exception e) {
			logger.error("创建节点出错，path：[{}],data:[{}],isPersistent:[{}]",path,data,isPersistent);
		}
	}
	
	public void testmodify(){
		String data = "hello world";
        try {
			client.setData().forPath("/zkTest", data.getBytes());
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	

}
